package com.atguigu.userprofile.utils.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.TaskInfo;
import com.atguigu.userprofile.bean.TaskTagRule;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.dao.TaskInfoDAO;
import com.atguigu.userprofile.dao.TaskTagRuleDAO;
import com.atguigu.userprofile.utils.MyPropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskSqlApp {
    public static void main(String[] args) {
        //注意 打包上传到集群运行的时候 不要带setMaster  带上这个setMaster单纯是为了在本地做测试的时候用
        SparkConf sparkConf = new SparkConf().setAppName("TaskSqlApp");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String dwDBName = properties.getProperty("data-warehouse.dbname");
        String upDBName = properties.getProperty("user-profile.dbname");


//        1.获取传进来的参数（TaskId（标签任务在mysql任务表中对应的id），BusiDate（业务日期））
        String taskId = args[0];
        String busiDate = args[1];
        System.out.println(taskId);
        System.out.println(busiDate);

//        2.根据TaskId获取task_info、tag_info、task_tag_rule这三张表的数据 利用jdbc去查询mysql中的数据
        TagInfo tagInfo = TagInfoDAO.getTagInfoWithTaskId(taskId);
        TaskInfo taskInfo = TaskInfoDAO.getTaskInfoWithTaskId(taskId);
        List<TaskTagRule> taskTagRuleList = TaskTagRuleDAO.getTaskTagRuleWithTaskId(taskId);

//        System.out.println(tagInfo);
//        System.out.println(taskInfo);
//        System.out.println(taskTagRuleList);

//        3.在tag_info中提取tag_code,tag_value_type,tag_name,在task_info中提取sql语句,在task_tag_rule获取标签值的映射关系
        String tableName = tagInfo.getTagCode().toLowerCase();

        String tagValueType = tagInfo.getTagValueType();
        String valueTypeSQL=null;
        if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagValueType)){
            valueTypeSQL = "bigint";
        }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagValueType)){
            valueTypeSQL = "decimal";
        }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagValueType)){
            valueTypeSQL = "string";
        }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagValueType)){
            valueTypeSQL = "string";
        }


        String tableComment = tagInfo.getTagName();

        String taskSql = taskInfo.getTaskSql();


//        4.动态拼接建表语句
        /**
         * create table if not exists tg_person_base_gender
         * (uid STRING,tag_value string)
         * partitioned by (dt string)
         * comment '性别'
         * row format delimited fields terminated by '\t'
         * location 'hdfs://hadoop102:8020/user_profile/user_profile220828/tg_person_base_gender'
         */

        String createTableSQL = " create table if not exists "+upDBName+"."+tableName+"\n" +
                "          (uid STRING,tag_value "+valueTypeSQL+")\n" +
                "         partitioned by (dt string)\n" +
                "         comment '"+tableComment+"'\n" +
                "         row format delimited fields terminated by '\\t'\n" +
                "         location '"+hdfsPath+"/"+upDBName+"/"+tableName+"'";

        System.out.println(createTableSQL);

//        5.动态拼接查询语句
        /**
         * select
         * uid,
         * case query_value when 'F' then '女' when 'M' then '男' when 'U' then '未知' end from  (
         * select
         * id as uid,
         * if(gender<>"",gender,"U") as query_value
         * from
         * dim_user_zip
         * where dt = '9999-12-31')duz
         */

        String querySQL = null;

        //如果查询出来规则表中有数据证明有四级标签，那么需要拼case when then  如果没有四级标签，则不需要拼
        if (taskTagRuleList.size()>0){

            List<String> whenThenSQLList = taskTagRuleList.stream().map(taskTagRule -> "when '" + taskTagRule.getQueryValue() + "' then '" + taskTagRule.getSubTagValue() + "'").collect(Collectors.toList());

            String whenThenSQL = StringUtils.join(whenThenSQLList, " ");
            System.out.println(whenThenSQL);

           querySQL = " select\n" +
                    "        uid,\n" +
                    "        case query_value "+whenThenSQL+" end as tag_value " +
                    "from  ("+taskSql+")duz";
        }else {
            querySQL = "select uid, query_value as tag_value from (" + taskSql + ")duz";
        }

        System.out.println(querySQL);

//        6.动态拼接插入语句
        /**
         * insert overwrite table tg_person_base_gender partition(dt='2020-06-14') select
         *  uid,
         *  case query_value when 'F' then '女'  when 'M' then '男'  when 'U' then '未知'   end as query_value
         *  from
         *  (select id as uid,if(gender<>"",gender,"U") as query_value from dim_user_zip where dt='9999-12-31')ti
         */
        String insertSQL = "insert overwrite table " +upDBName+"."+ tableName + " partition(dt='" + busiDate + "') " + querySQL;

        System.out.println("插入语句");
        System.out.println(insertSQL);
//        7.执行sql
        sparkSession.sql("use " + dwDBName);
        sparkSession.sql(createTableSQL);
        sparkSession.sql(insertSQL);

    }
}
